package com.intct.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.producer.ProducerConfig;

/**
 * @author gufg
 * @since 2025-09-17 15:02
 */
public class KafkaUtil {
    /**
     * 配置KafkaSink
     * @param parameterTool
     * @param topic
     * @param tranIdPrefix
     * @return
     */
    public static KafkaSink<String> getKafaSink(ParameterTool parameterTool, String topic, String tranIdPrefix) {
        return KafkaSink.<String>builder()
                .setBootstrapServers(parameterTool.get("kafka.sink.bootstrap.servers", "node1:9092"))
                .setRecordSerializer(KafkaRecordSerializationSchema.<String>builder()
                        .setTopic(topic)
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build())
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                .setTransactionalIdPrefix(tranIdPrefix)
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, parameterTool.get("kafka.sink.transaction.timeout.config", "60000"))
                .build();
    }

    /**
     * Kafka源配置
     * @param parameterTool
     * @param topic
     * @return
     */
    public static KafkaSource<String> getKafkaSource(ParameterTool parameterTool, String topic) {
        return KafkaSource.<String>builder()
                .setBootstrapServers(parameterTool.get("kafka.sink.bootstrap.servers", ""))
                .setTopics(topic)
                .setValueOnlyDeserializer(new SimpleStringSchema())
                // 主要是讲课使用earliest，生产环境使用commmit offset
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setGroupId("dim-main-app-group-1")
                .build();
    }
}
